package com.yh.consumer;

import com.rabbitmq.client.Channel;
import com.yh.config.OtherSource;
import com.yh.dao.MovieDao;
import com.yh.entity.Movie;
import com.yh.pojo.MqMessage;
import com.yh.service.ElasticsearchService;
import com.yh.utils.Constants;
import com.yh.utils.ImgUrl;
import io.minio.MinioClient;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.log4j.Logger;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

/**
 * @author yuhuan
 * @date 2022/01/13
 */
@Service
@RabbitListener(queues = Constants.RabbitMqData.PREHEAT_QUEUE_NAME, ackMode = "MANUAL")
public class MovieModuleConsumerBase extends BaseMqConsumer {
    /**
     * 日志
     */
    private static final Logger logger = Logger.getLogger(MovieModuleConsumerBase.class);

    @Autowired
    private MovieDao movieDao;
    @DubboReference(check = false)
    private ElasticsearchService esService;
    @Autowired
    private MinioClient minioClient;
    @Autowired
    private OtherSource otherSource;

    @RabbitHandler
    public void receiveMoviePreheatMessage(String movieId, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        logger.info("处理预热请求，影片ID：" + movieId);
        Movie movie = movieDao.findMovieById(Integer.valueOf(movieId));
        if (ObjectUtils.isEmpty(movie)) {
            return;
        }
        // 更换影片的海报地址
        movie = ImgUrl.movieImageHandler(minioClient, otherSource, movie);
        // 预热影片到搜索引擎
        moviePreheatToElasticsearch(movie);
        // 消息确认
        messageConfirmation(channel, deliveryTag);
    }

    @RabbitHandler
    public void receiveMovieEntityMessage(Movie movie, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        logger.info("接收到新的影片信息：" + movie);
        if (ObjectUtils.isEmpty(movie)) {
            return;
        }
        // 更换影片的海报地址
        movie = ImgUrl.movieImageHandler(minioClient, otherSource, movie);
        // 添加影片到搜索引擎
        moviePreheatToElasticsearch(movie);
        // 消息确认
        messageConfirmation(channel, deliveryTag);
    }

    @RabbitHandler
    public void receiveMovieMessage(MqMessage message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        logger.info("接收到新的指令：" + message);
        if (ObjectUtils.isEmpty(message)) {
            return;
        }
        String type = message.getType();
        if (type.equals(Constants.MqMessageData.DEL_MESSAGE)) {
            Integer movieId = (Integer) message.getData();
            // 移除搜索引擎中的影片信息
            esService.deleteDocument(movieId);
        }
        // 消息确认
        messageConfirmation(channel, deliveryTag);
    }

    @RabbitHandler(isDefault = true)
    /** 备用消费者（消费现有消费者无法接收的任何类型的数据） */
    public void receiveMessage(Object msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        logger.info("接收到消息：" + msg);
        // 消息确认
        messageConfirmation(channel, deliveryTag);
    }

    /**
     * 将影片的信息预热到搜索引擎
     */
    private void moviePreheatToElasticsearch(Movie movie) {
        esService.createDocument(movie);
    }
}
